﻿#region LicenseHeader

// Copyright 2012 The Trustees of Indiana University.  All rights reserved.
// 
// The Indiana University licenses this file to you under the Apache License, 
// Version 2.0 (the "License"); you may not use this file except in compliance 
// with the License.  You may obtain a copy of the License at 
// 
// http://www.apache.org/licenses/LICENSE-2.0
// 
// Unless required by applicable law or agreed to in writing, software 
// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
//  WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  
// See the License for the specific language governing permissions and 
// limitations under the License.
// 
// @author Thilina Gunarathne (tgunarat@indiana.edu)

#endregion

using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Net;
using System.ServiceModel;
using System.Threading;
using AzureMRCore;
using AzureMRCore.DataCommunication;
using AzureMRCore.DataModel;
using AzureMRCore.Drivers;
using AzureMRCore.IterativeDataCache;
using AzureMRCore.MapRed;
using AzureMRCore.Primitives;
using Microsoft.WindowsAzure;
using Microsoft.WindowsAzure.Diagnostics;
using Microsoft.WindowsAzure.ServiceRuntime;
using Microsoft.WindowsAzure.StorageClient;

namespace Twister4AzureWorker
{
    public class MRWorkerRole : RoleEntryPoint
    {
        /// <summary>
        ///   Channel factory for inter-role notifications.
        /// </summary>
        private static ChannelFactory<IDataXferService> _factory;

        /// <summary>
        ///   ServiceHost object for internal and external endpoints.
        /// </summary>
        private ServiceHost _serviceHost;

        protected CloudStorageAccount StorageAccount;

        public override sealed bool OnStart()
        {
            ServicePointManager.UseNagleAlgorithm = false;
            ServicePointManager.Expect100Continue = false;
            //ServicePointManager.CheckCertificateRevocationList = false;
            ServicePointManager.DefaultConnectionLimit = 40;

            CloudStorageAccount.SetConfigurationSettingPublisher(
                (configName, configSetter) => { configSetter(RoleEnvironment.GetConfigurationSettingValue(configName)); });

            StorageAccount = CloudStorageAccount.FromConfigurationSetting("DataConnectionString");

            DiagnosticMonitorConfiguration config = DiagnosticMonitor.GetDefaultInitialConfiguration();
            config.PerformanceCounters.DataSources.Add(
                new PerformanceCounterConfiguration()
                    {
                        CounterSpecifier = @"\Processor(_Total)\% Processor Time",
                        SampleRate = TimeSpan.FromSeconds(60)
                    });
            config.PerformanceCounters.DataSources.Add(
                new PerformanceCounterConfiguration()
                    {
                        CounterSpecifier = @"\Memory\Available MBytes",
                        SampleRate = TimeSpan.FromSeconds(60)
                    });
            config.PerformanceCounters.DataSources.Add(
                new PerformanceCounterConfiguration()
                    {
                        CounterSpecifier = @"\Memory\Committed Bytes",
                        SampleRate = TimeSpan.FromSeconds(60)
                    });

            config.Logs.ScheduledTransferPeriod = TimeSpan.FromMinutes(5);
            config.PerformanceCounters.ScheduledTransferPeriod = TimeSpan.FromMinutes(10);
            DiagnosticMonitor.Start("DiagnosticsConnectionString", config);

            // For information on handling configuration changes
            // see the MSDN topic at http://go.microsoft.com/fwlink/?LinkId=166357.
            RoleEnvironment.Changing += RoleEnvironmentChanging;

            return base.OnStart();
        }

        public override sealed void Run()
        {
            try
            {
                StartDataXferService(10);
                var conf = MRConf.GetInstance();
                ParseMRConfig(conf);

                RunMapRed(StorageAccount, conf);
            }
            catch (Exception e)
            {
                LogError(e, "MRWorkerRole", e.Message);
            }
        }

        protected void RunMapRed(CloudStorageAccount storageAccount, MRConf mrConf)
        {
            //Create or get references to the Map & Reduce Scheduling queues
            CloudQueueClient queueStorage = storageAccount.CreateCloudQueueClient();
            CloudQueue mapSchedQueue = queueStorage.GetQueueReference(mrConf.MapSchedQueueName);
            mapSchedQueue.CreateIfNotExist();

            CloudQueue reduceSchedQueue = queueStorage.GetQueueReference(mrConf.ReduceSchedQueueName);
            reduceSchedQueue.CreateIfNotExist();

            var dataCache = new DataCache(storageAccount);
            var mapTaskHistories = new Dictionary<string, List<MapTaskHistory>>();

            /* var bCastReceiverDeamon = new BroadcastReceiverDeamon(storageAccount);
            var bCastReceiverThread = new Thread(bCastReceiverDeamon.RunBroadcastReceiver);
            bCastReceiverThread.Start();*/
            mrConf.BCastReceiver = BCastServiceReceiver.GetInstance();
            mrConf.Factory = _factory;

            while (true)
            {
                // start the map worker threads
                var mapThreads = new Thread[mrConf.MapWorkersPerInstance];
                for (int i = 0; i < mrConf.MapWorkersPerInstance; i++)
                {
                    var mapWorker = new MapWorker(storageAccount, mapSchedQueue, mrConf, dataCache, mapTaskHistories);
                    mapThreads[i] = new Thread(mapWorker.RunMapWorker);
                    mapThreads[i].Start();
                }

                //TODO start the data transfer worker
                //start the reduce worker threads
                var reduceThreads = new Thread[mrConf.ReduceWorkersPerInstance];

                for (int i = 0; i < mrConf.ReduceWorkersPerInstance; i++)
                {
                    var reduceWorker = new ReduceWorker(storageAccount, reduceSchedQueue, mrConf, dataCache);
                    reduceThreads[i] = new Thread(reduceWorker.RunReduce);
                    reduceThreads[i].Start();
                }


                //Wait for all the map worker threads to finish
                for (int i = 0; i < mrConf.MapWorkersPerInstance; i++)
                {
                    mapThreads[i].Join();
                }

                //Wait for all the reduce worker threads to finish
                for (int i = 0; i < mrConf.ReduceWorkersPerInstance; i++)
                {
                    reduceThreads[i].Join();
                }
            }
        }

        /// <summary>
        ///   Starts the service host object for the internal and external endpoints of the chat service.
        /// </summary>
        /// <param name="retries"> Specifies the number of retries to start the service in case of failure. </param>
        private void StartDataXferService(int retries)
        {
            // recycle the role if host cannot be started 
            // after the specified number of retries
            if (retries == 0)
            {
                RoleEnvironment.RequestRecycle();
                return;
            }

            _serviceHost = new ServiceHost(typeof(DataXferService));

            // Recover the service in case of failure. 
            // Log the fault and attempt to restart the service host. 
            _serviceHost.Faulted += (sender, e) =>
            {
                Trace.TraceError(
                    "Host fault occured. Aborting and restarting the host. Retry count: {0}",
                    retries);
                _serviceHost.Abort();
                StartDataXferService(--retries);
            };

            // use NetTcpBinding with no security
            var binding = new NetTcpBinding(SecurityMode.None)
            {
                MaxConnections = 128,
                MaxReceivedMessageSize = int.MaxValue,
                ReaderQuotas = { MaxArrayLength = int.MaxValue, MaxBytesPerRead = 12800 }
            };
            // binding.TransferMode = TransferMode.Streamed;
            binding.SendTimeout.Add(new TimeSpan(0, 0, 5, 0));
            binding.CloseTimeout = TimeSpan.MaxValue;

            // define an internal endpoint for inter-role traffic
            RoleInstanceEndpoint internalEndPoint =
                RoleEnvironment.CurrentRoleInstance.InstanceEndpoints["NotificationService"];
            _serviceHost.AddServiceEndpoint(
                typeof(IDataXferService),
                binding,
                String.Format("net.tcp://{0}/NotificationService", internalEndPoint.IPEndpoint));

            // create channel factory for inter-role communication
            _factory = new ChannelFactory<IDataXferService>(binding);

            try
            {
                _serviceHost.Open();
                //   Trace.TraceInformation("Chat service host started successfully.");
            }
            catch (TimeoutException timeoutException)
            {
                Trace.TraceError("The service operation timed out. {0}",
                                 timeoutException.Message);
            }
            catch (CommunicationException communicationException)
            {
                Trace.TraceError("Could not start chat service host. {0}",
                                 communicationException.Message);
            }
        }

        private void ParseMRConfig(MRConf conf)
        {
            string stringConfig;
            int intConfig;
            if ((stringConfig = GetStringConfig("TwisterMapQName")) != null)
            {
                conf.MapSchedQueueName = stringConfig;
            }

            if ((stringConfig = GetStringConfig("TwisterReduceQName")) != null)
            {
                conf.ReduceSchedQueueName = stringConfig;
            }

            if ((intConfig = GetIntConfig("TwisterMapTimeout")) >= 0)
            {
                conf.MapTimeOut = intConfig;
            }

            if ((intConfig = GetIntConfig("TwisterReduceTimeout")) >= 0)
            {
                conf.ReduceTimeOut = intConfig;
            }

            if ((intConfig = GetIntConfig("TwisterNumMapWorkersPerInstance")) >= 0)
            {
                conf.MapWorkersPerInstance = intConfig;
            }

            if ((intConfig = GetIntConfig("TwisterNumReduceWorkersPerInstance")) >= 0)
            {
                conf.ReduceWorkersPerInstance = intConfig;
            }

            if ((intConfig = GetIntConfig("TwisterMapQPollingInterval")) >= 0)
            {
                conf.MapQPollingInterval = intConfig;
            }

            if ((intConfig = GetIntConfig("TwisterReduceQPollingInterval")) >= 0)
            {
                conf.RedQPollingInterval = intConfig;
            }

            if ((intConfig = GetIntConfig("ParallelBCastMax")) >= 0)
            {
                conf.ParallelBCastMax = intConfig;
            }

            stringConfig = GetStringConfig("TwisterMRDrivers");
            if (String.IsNullOrWhiteSpace(stringConfig))
            {
                throw new Exception("No MR drivers specified.");
            }
            string[] mrDrivers = stringConfig.Split(';');
            foreach (string mrDriver in mrDrivers)
            {
                Type driverType = Type.GetType(mrDriver);
                if (driverType != null)
                {
                    var driver = (MRDriver)Activator.CreateInstance(driverType);
                    object configOut;
                    driver.ConfigMapRed(conf, StorageAccount, out configOut);
                    driver.ConfigOut = configOut;
                    conf.MRDrivers.Add(driver.Name, driver);
                    Trace.WriteLine("Added MR Application : " + mrDriver);
                }
                else
                {
                    LogError(null, "MRWorkerRole", mrDriver + " app type cannot be resolved.");
                }
            }
        }

        private static string GetStringConfig(string configName)
        {
            try
            {
                string configString = RoleEnvironment.GetConfigurationSettingValue(configName);
                if (!String.IsNullOrWhiteSpace(configString))
                {
                    return configString;
                }
            }
            catch (RoleEnvironmentException e)
            {
            }
            return null;
        }

        private static int GetIntConfig(string configName)
        {
            try
            {
                string configString = RoleEnvironment.GetConfigurationSettingValue(configName);
                if (!String.IsNullOrWhiteSpace(configString))
                {
                    int configVal;
                    if (Int32.TryParse(configString, out configVal))
                    {
                        return configVal;
                    }
                }
            }
            catch (RoleEnvironmentException e)
            {
            }
            return -1;
        }

        private void RoleEnvironmentChanging(object sender, RoleEnvironmentChangingEventArgs e)
        {
            // If a configuration setting is changing
            if (e.Changes.Any(change => change is RoleEnvironmentConfigurationSettingChange))
            {
                // Set e.Cancel to true to restart this role instance
                e.Cancel = true;
            }
        }

        private void LogError(Exception e, string source, string message)
        {
            CloudTableClient tableClient = StorageAccount.CreateCloudTableClient();
            var errorContext = new ErrorDataContext(StorageAccount.TableEndpoint.ToString(),
                                                    StorageAccount.Credentials, "AzureMRErrors");
            tableClient.CreateTableIfNotExist(errorContext.Name);
            var error = new ErrorDataModel(e, source, message);
            errorContext.AddObject(errorContext.Name, error);
            errorContext.SaveChangesWithRetries();
        }
    }
}